package cn.sunjinxin.product.application.service;

import cn.sunjinxin.product.client.ProductService;
import cn.sunjinxin.product.client.request.InventoryRequest;
import cn.sunjinxin.product.domain.event.InventoryEvent;
import cn.sunjinxin.savior.core.common.Result;
import cn.sunjinxin.savior.event.context.EventContext;
import cn.sunjinxin.savior.event.control.Eventer;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.commons.collections4.CollectionUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisCommands;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * @author issavior
 */
@Service
public class ProductServiceImpl implements ProductService {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private JdbcTemplate jdbcTemplate;

    private static final RateLimiter LIMITER = RateLimiter.create(2.0);

    @Override
    public Result<Void> preInventory(InventoryRequest request) {

        LIMITER.acquire();

        List<Object> preInventoryResult = redisTemplate.executePipelined((RedisCallback<String>) connection -> {
            RedisCommands commands = connection.commands();
            Long inventory = commands.decrBy(("inventory" + request.getProductSkuId()).getBytes(), request.getBuyQuantity());
            if (inventory == null || inventory < 0) {
                commands.incrBy(("inventory" + request.getProductSkuId()).getBytes(), request.getBuyQuantity());
                return "fail:库存不足";
            }
            if (null != commands.hGet(("record" + request.getProductSkuId()).getBytes(), request.getUniqueId().getBytes())) {
                return "fail:不可重复扣减";
            }
            commands.hSet((("record" + request.getProductSkuId())).getBytes(), request.getUniqueId().getBytes(), (new Date().getTime() + "扣减" + request.getBuyQuantity()).getBytes());
            return null;
        });

        if (CollectionUtils.isNotEmpty(preInventoryResult)) {
            return Result.failed(preInventoryResult.get(0).toString());
        }

        Eventer.SYNC.publish(EventContext.<InventoryEvent, InventoryRequest>builder()
                .eventId(String.valueOf(request.getProductSkuId()))
                .eventType(InventoryEvent.PRE_INVENTORY)
                .request(request)
                .build());
        return null;
    }

    @Override
    @Transactional
    public Result<Void> updateInventory(InventoryRequest request) {
        List<Object> results = redisTemplate.execute(new SessionCallback<List<Object>>() {
            @Override
            public List<Object> execute(@NotNull RedisOperations operations) throws DataAccessException {
                @SuppressWarnings("unchecked")
                RedisOperations<String, Object> operation = operations;
                operation.watch("inventory" + request.getProductSkuId()); // 监视指定的键
                jdbcTemplate.execute("update table set total_inventory = total_inventory + ? where productSkuId = #{productSkuId}");
                // 开启事务
                operation.multi();
                operation.opsForValue().increment("inventory" + request.getProductSkuId(), request.getBuyQuantity());
                // 提交事务（若键被修改，返回 null）
                return operation.exec();
            }
        });

        // 判断事务是否成功
        if (results != null && !results.isEmpty()) {
            return Result.ok();
        }

        throw new RuntimeException("库存扣减失败");
    }

    @Override
    public Result<Void> consistentInventory(InventoryRequest request) {
        List<Object> results = redisTemplate.execute(new SessionCallback<List<Object>>() {
            @Override
            public List<Object> execute(@NotNull RedisOperations operations) throws DataAccessException {
                @SuppressWarnings("unchecked")
                RedisOperations<String, Object> operation = operations;
                operation.watch("inventory" + request.getProductSkuId()); // 监视指定的键

                Map<String, Object> list = jdbcTemplate.queryForMap("select * from tb_inventory where productSkuId = #{productSkuId} for update");
                List<Map<String, Object>> listTasks = jdbcTemplate.queryForList("select * from tb_inventory_task where productSkuId = #{productSkuId} and status = '待执行' for update");
                int total = (int) list.get("total_inventory") + listTasks.size();
                Map<Object, Object> entries = redisTemplate.opsForHash().entries("record" + request.getProductSkuId());
                int size = entries.values().size();

                if (total == size) {
                    return null;
                }
                if (total < size) {
                    throw new RuntimeException("超卖预警");
                }

                // 开启事务
                operation.multi();
                operation.opsForValue().increment("inventory" + request.getProductSkuId(), request.getBuyQuantity());
                // 提交事务（若键被修改，返回 null）
                return operation.exec();
            }
        });

        // 判断事务是否成功
        if (results != null && !results.isEmpty()) {
            return Result.ok();
        }

        return Result.failed("对账失败");

    }
}
